package com.itheima.safe.mq;

import com.alibaba.fastjson.JSON;
import com.itheima.common.pojo.VerifyMessage;
import com.itheima.dfs.feign.DfsFeign;
import com.itheima.safe.service.AliyunSafeService;
import com.itheima.safe.util.MatchUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author：shenkunlin
 * @date： 2021/8/26 10:23
 * @Description：
 ***/
@Component
public class VerifyConsumer {

    @Autowired
    private AliyunSafeService aliyunSafeService;

    @Autowired
    private DfsFeign dfsFeign;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 内容安全检测
     * content:需要检测的内容
     * backqueue:需要将检测结果返回的队列
     * attach:附加数据（用于校验）
     */
    @KafkaListener(topics = "checkfortext",groupId = "safegroup")
    public void verifySafe(ConsumerRecord<?,?> record) throws Exception {
        //获取内容
        String content = record.value().toString();

        //1）将content转换成VerifyMessage
        VerifyMessage verifyMessage = JSON.parseObject(content,VerifyMessage.class);

        //2）将VerifyMessage中的content拿到aliyun检测
        String verifyContent = verifyMessage.getContent();

        //内容安全检测
        boolean result = true;
        //将整个文本进行安全检测(包含url地址，都会被视为广告)
        if(!aliyunSafeService.verifyText(Collections.singletonList(MatchUtil.replateUrl(verifyContent)))){
            result=false;
        }
        //匹配文本中的图片进行安全检测
        List<String> images = MatchUtil.matchImages(content);
        //文件的字节数组-fastdfs
        if(images!=null && images.size()>0){
            //批量下载
            List<byte[]> iamgesBytes = dfsFeign.download(images);
            if(!aliyunSafeService.verifyImages(iamgesBytes)){
                //无垃圾内容
                result=false;
            }
        }
        //将结果返回出去->向Kafka发送消息
        //3）将检测的结果和元数据返回到指定队列（VerifyMessage.backqueue）[wmnews_save_queue]
        Map<String,Object> resultMap = new HashMap<String,Object>();
        resultMap.put("attach",verifyMessage.getAttach());
        resultMap.put("result",result);
        kafkaTemplate.send(verifyMessage.getBackqueue(),JSON.toJSONString(resultMap));
    }
}
